# -*- coding: utf-8 -*-
# ------------------------------------------------------------------------
#  Copyright by KNIME AG, Zurich, Switzerland
#  Website: http://www.knime.com; Email: contact@knime.com
#
#  This program is free software; you can redistribute it and/or modify
#  it under the terms of the GNU General Public License, Version 3, as
#  published by the Free Software Foundation.
#
#  This program is distributed in the hope that it will be useful, but
#  WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, see <http://www.gnu.org/licenses>.
#
#  Additional permission under GNU GPL version 3 section 7:
#
#  KNIME interoperates with ECLIPSE solely via ECLIPSE's plug-in APIs.
#  Hence, KNIME and ECLIPSE are both independent programs and are not
#  derived from each other. Should, however, the interpretation of the
#  GNU GPL Version 3 ("License") under any applicable laws result in
#  KNIME and ECLIPSE being a combined program, KNIME AG herewith grants
#  you the additional permission to use and propagate KNIME together with
#  ECLIPSE with only the license terms in place for ECLIPSE applying to
#  ECLIPSE and the GNU GPL Version 3 applying for KNIME, provided the
#  license terms of ECLIPSE themselves allow for the respective use and
#  propagation of ECLIPSE together with KNIME.
#
#  Additional permission relating to nodes for KNIME that extend the Node
#  Extension (and in particular that are based on subclasses of NodeModel,
#  NodeDialog, and NodeView) and that only interoperate with KNIME through
#  standard APIs ("Nodes"):
#  Nodes are deemed to be separate and independent programs and to not be
#  covered works.  Notwithstanding anything to the contrary in the
#  License, the License does not apply to Nodes, you are not required to
#  license Nodes under the License, and you are granted a license to
#  prepare and propagate Nodes, in each case even if such Nodes are
#  propagated with or for interoperation with KNIME.  The owner of a Node
#  may freely choose the license terms applicable to such Node, including
#  when such Node is propagated with or for interoperation with KNIME.
# ------------------------------------------------------------------------

"""
@author Marcel Wiedenmann, KNIME GmbH, Konstanz, Germany
@author Christian Dietz, KNIME GmbH, Konstanz, Germany
"""

import abc
import threading
from concurrent import futures
from queue import Full

from PythonUtils import invoke_safely


class AbstractMessageLoop(object):
    __metaclass__ = abc.ABCMeta

    @staticmethod
    def _clear_queue_and_put_message(queue, message):
        while True:
            with queue.mutex:
                queue.queue.clear()
            try:
                queue.put_nowait(message)
            except Full:
                continue
            else:
                break

    def __init__(self, monitor):
        self._monitor = monitor
        self._executor = futures.ThreadPoolExecutor(1)
        self._is_running = False
        self._is_running_lock = threading.RLock()
        self._is_closed_or_closing = False
        self._is_closed_or_closing_lock = threading.RLock()
        self._exception_during_close = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    @abc.abstractmethod
    def _loop(self):
        raise NotImplementedError()

    @property
    def is_running(self):
        with self._is_running_lock:
            return self._is_running

    def start(self):
        with self._is_running_lock:
            if not self._is_running:
                with self._is_closed_or_closing_lock:
                    if self._is_closed_or_closing:
                        raise RuntimeError("Cannot restart a closed message loop.")
                    else:
                        self._is_running = True
                        self._executor.submit(self._do_loop)

    def close(self):
        with self._is_running_lock:
            if self._is_running:
                self._is_running = False
                with self._is_closed_or_closing_lock:
                    self._is_closed_or_closing = True
                invoke_safely(None, lambda e: e.shutdown(wait=False), self._executor)
                self._close()
            elif self._exception_during_close is not None:
                raise self._exception_during_close

    def _close(self):
        pass  # No-op by default.

    def _do_loop(self):
        try:
            self._loop()
        except BaseException as ex:
            self._monitor.report_exception(ex, message=self.__class__.__name__ + " terminated.")
            raise
        finally:
            try:
                self.close()
            except BaseException as ex:
                self._exception_during_close = ex
